
package org.hbhk.aili.job.configuration;

import org.hbhk.aili.job.event.MyElasticJobListener;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.annotation.Autowired;

import com.dangdang.ddframe.job.api.ElasticJob;
import com.dangdang.ddframe.job.api.dataflow.DataflowJob;
import com.dangdang.ddframe.job.api.script.ScriptJob;
import com.dangdang.ddframe.job.api.simple.SimpleJob;
import com.dangdang.ddframe.job.config.JobCoreConfiguration;
import com.dangdang.ddframe.job.config.JobTypeConfiguration;
import com.dangdang.ddframe.job.config.dataflow.DataflowJobConfiguration;
import com.dangdang.ddframe.job.config.script.ScriptJobConfiguration;
import com.dangdang.ddframe.job.config.simple.SimpleJobConfiguration;
import com.dangdang.ddframe.job.event.JobEventConfiguration;
import com.dangdang.ddframe.job.executor.handler.JobProperties;
import com.dangdang.ddframe.job.lite.api.JobScheduler;
import com.dangdang.ddframe.job.lite.api.listener.ElasticJobListener;
import com.dangdang.ddframe.job.lite.config.LiteJobConfiguration;
import com.dangdang.ddframe.job.reg.base.CoordinatorRegistryCenter;
import com.google.common.base.Optional;


public class ElasticScheduler implements InitializingBean {
	private ElasticConfig elasticConfig;
	private ElasticJob elasticJob;
	@Autowired
	private CoordinatorRegistryCenter registryCenter;
	@Autowired(required = false)
	private JobEventConfiguration jobEventConfiguration;

	@Override
	public void afterPropertiesSet() throws Exception {
		String jobName = elasticConfig.getName();
		String corn = elasticConfig.getCron();
		int shardingCount = elasticConfig.getShardingCount();

		JobCoreConfiguration jobConfig = JobCoreConfiguration.newBuilder(jobName, corn, shardingCount)
				.description(elasticConfig.getDescription()).failover(elasticConfig.isFailover())
				.misfire(elasticConfig.isMisfire()).shardingItemParameters(elasticConfig.getShardingItemParameters())
				.jobParameter(elasticConfig.getJobParameter())
				.jobProperties(JobProperties.JobPropertiesEnum.JOB_EXCEPTION_HANDLER.getKey(),
						elasticConfig.getJobExceptionHandler().getCanonicalName())
				.jobProperties(JobProperties.JobPropertiesEnum.EXECUTOR_SERVICE_HANDLER.getKey(),
						elasticConfig.getExecutorServiceHandler().getCanonicalName())
				.build();
		if (null != jobEventConfiguration) {
			new SpringJobScheduler(registryCenter, buildJobConfig(jobConfig), jobEventConfiguration, elasticJob,
					new MyElasticJobListener()).init();
		} else {
			new SpringJobScheduler(registryCenter, buildJobConfig(jobConfig), elasticJob, new MyElasticJobListener())
					.init();
		}

	}

	LiteJobConfiguration buildJobConfig(JobCoreConfiguration jobConfig) {
		JobTypeConfiguration typeConfiguration = null;
		String jobClass = elasticJob.getClass().getCanonicalName();
		if (elasticJob instanceof SimpleJob) {
			typeConfiguration = new SimpleJobConfiguration(jobConfig, jobClass);
		} else if (elasticJob instanceof DataflowJob) {
			typeConfiguration = new DataflowJobConfiguration(jobConfig, jobClass, elasticConfig.isStreamingProcess());
		} else if (elasticJob instanceof ScriptJob) {
			typeConfiguration = new ScriptJobConfiguration(jobConfig, elasticConfig.getScriptCommandLine());
		}
		return LiteJobConfiguration.newBuilder(typeConfiguration).monitorExecution(elasticConfig.isMonitorExecution())
				.monitorPort(elasticConfig.getMonitorPort()).maxTimeDiffSeconds(elasticConfig.getMaxTimeDiffSeconds())
				.jobShardingStrategyClass(elasticConfig.getJobShardingStrategyClass())
				.reconcileIntervalMinutes(elasticConfig.getReconcileIntervalMinutes())
				.disabled(elasticConfig.isDisabled()).overwrite(elasticConfig.isOverwrite()).build();
	}

	static class SpringJobScheduler extends JobScheduler {
		private final ElasticJob elasticJob;

		public SpringJobScheduler(CoordinatorRegistryCenter regCenter, LiteJobConfiguration liteJobConfig,
				ElasticJob elasticJob, ElasticJobListener... elasticJobListeners) {
			super(regCenter, liteJobConfig, elasticJobListeners);
			this.elasticJob = elasticJob;
		}

		public SpringJobScheduler(CoordinatorRegistryCenter regCenter, LiteJobConfiguration liteJobConfig,
				JobEventConfiguration jobEventConfiguration, ElasticJob elasticJob,
				ElasticJobListener... elasticJobListeners) {
			super(regCenter, liteJobConfig, jobEventConfiguration, elasticJobListeners);
			this.elasticJob = elasticJob;
		}

		@Override
		protected Optional<ElasticJob> createElasticJobInstance() {
			return Optional.of(elasticJob);
		}
	}

	public ElasticConfig getElasticConfig() {
		return elasticConfig;
	}

	public void setElasticConfig(ElasticConfig elasticConfig) {
		this.elasticConfig = elasticConfig;
	}

	public ElasticJob getElasticJob() {
		return elasticJob;
	}

	public void setElasticJob(ElasticJob elasticJob) {
		this.elasticJob = elasticJob;
	}

}
